package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.execption.GuiguException;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.model.album.AlbumInfo;
import com.atguigu.tingshu.model.album.BaseCategoryView;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.search.executor.ExpireThreadExecutor;
import com.atguigu.tingshu.search.factory.ScheduleTaskThreadPoolFactory;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.runnable.RebuildBloomFilterRunnable;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import com.google.common.collect.Lists;
import java.util.*;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Supplier;
import java.util.stream.Collectors;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.search.repository.AlbumInfoIndexRepository;
import com.atguigu.tingshu.search.service.ItemService;
import com.google.common.hash.BloomFilter;
import jakarta.annotation.PostConstruct;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;

import org.chb.cache.service.CacheOpsService;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.core.script.DefaultRedisScript;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class ItemServiceImpl implements ItemService {

    @Autowired
    private AlbumInfoIndexRepository albumInfoIndexRepository;

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;

    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    private UserInfoFeignClient userInfoFeignClient;

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private StringRedisTemplate redisTemplate;

    @Autowired
    private RBloomFilter rBloomFilter;

    @Autowired
    private RedissonClient redissonClient;

    @Autowired
    private ItemServiceImpl itemServiceImpl;
    @Autowired
    private CacheOpsService cacheOpsService;


    ExecutorService executorService = Executors.newFixedThreadPool(4);
    ThreadLocal<String> reentrantLockTokenThreadLocal = new ThreadLocal<>();
    Map<Long,Map<String,Object>> localCache = new HashMap<>(); //定义一个本地Map缓存

    BloomFilter<Long> longBloomFilter = null;

    @PostConstruct  //spring在创建ItemServiceImpl Bean对象的时候，在生命周期走到初始化前这个阶段就会回来调该方法
    public void initRebuildBloomFilter(){
//        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
//        scheduledExecutorService.scheduleWithFixedDelay(new RebuildBloomFilterRunnable(redissonClient,redisTemplate,itemServiceImpl),0,7,TimeUnit.DAYS);
//        scheduledExecutorService.schedule(new RebuildBloomFilterRunnable(redissonClient,redisTemplate,itemServiceImpl),10,TimeUnit.DAYS);

        //测试使用的
//        ScheduleTaskThreadPoolFactory instance = ScheduleTaskThreadPoolFactory.getInstance();
//        instance.execute(new RebuildBloomFilterRunnable(redissonClient,redisTemplate,itemServiceImpl),10l,TimeUnit.SECONDS);


        // 从当前开始算 7天之后的凌晨2点执行第一次 --线上使用的
//        instance.execute(new RebuildBloomFilterRunnable(redissonClient,redisTemplate,itemServiceImpl),10l,TimeUnit.SECONDS);
        ScheduleTaskThreadPoolFactory instance = ScheduleTaskThreadPoolFactory.getInstance();
        Long taskFirstTime = instance.diffTime(System.currentTimeMillis());
        instance.execute(new RebuildBloomFilterRunnable(redissonClient,redisTemplate,itemServiceImpl),taskFirstTime,TimeUnit.MILLISECONDS);
    }

//    public void initLocalBloomFilter(){
//        //1. 创建布隆过滤器
//        Funnel<Long> longFunnel = Funnels.longFunnel();
//        longBloomFilter = BloomFilter.create(longFunnel,1000000,0.01);
//
//
//        //将元素放到布隆过滤器中
//        List<Long> albumInfoIdList = getAlbumInfoIdList();
//        albumInfoIdList.stream().forEach(albumId -> {
//            longBloomFilter.put(albumId);
//        });
//        log.info("本地布隆初始化完毕，且布隆中的元素个数：{}",longBloomFilter.approximateElementCount());
//    }

    /**
     * 1.数据从哪来 接着到哪里去
     * 2.读写模型
     * 3.请求响应模型
     * <p>
     * 如果单线程远程查询4次耗时  例如：第一次：322ms 第二次：87ms 第三次：62ms 第四次：56ms
     * OpenFeign底层有缓存机制：只有在调用方法去第一次调用时才会找注册中心要数据（发送HTTP请求）接着将被调用方的信息缓存到了本地Map中，
     * 因此当调用方后续在给被调用发轻轻时，直接从map中获取被调用方的信息
     * <p>
     * 异步和快没有直接的关系，异步主要让线程压榨cpu，不让cpu空闲一直干活
     * @param albumId
     */
    @SneakyThrows
    @Override
    public void albumOnSale(Long albumId)  {
        CountDownLatch countDownLatch = new CountDownLatch(4);

        //1.创建文档对象
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();

        long startTime = System.currentTimeMillis();

        //2.给albumInfoIndex属性赋值

        Future<Long> future = executorService.submit(new Callable<Long>() {
            @Override
            public Long call() throws Exception {
                //2.1 远程查询专辑基本信息

                Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfoAndAttrValue(albumId);

                AlbumInfo albumInfoData = albumInfoResult.getData();
                if (albumInfoData == null) {
                    throw new GuiguException(201, "远程调用专辑微服务获取专辑信息失败");
                }

                albumInfoIndex.setId(albumInfoData.getId()); //专辑id
                albumInfoIndex.setAlbumTitle(albumInfoData.getAlbumTitle()); //专辑标签
                albumInfoIndex.setAlbumIntro(albumInfoData.getAlbumIntro());//专辑简介

                albumInfoIndex.setCoverUrl(albumInfoData.getCoverUrl()); //专辑封面
                albumInfoIndex.setIncludeTrackCount(albumInfoData.getIncludeTrackCount());
                albumInfoIndex.setIsFinished(albumInfoData.getIsFinished().toString());
                albumInfoIndex.setPayType(albumInfoData.getPayType());
                albumInfoIndex.setCreateTime(new Date());

                List<AttributeValueIndex> attributeValueIndexs = albumInfoData.getAlbumAttributeValueVoList().stream()
                        .map(albumAttributeValue -> {
                            AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                            attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId());
                            attributeValueIndex.setValueId(albumAttributeValue.getValueId());
                            return attributeValueIndex;

                        }).collect(Collectors.toList());


                albumInfoIndex.setAttributeValueIndexList(Lists.newArrayList(attributeValueIndexs));
                countDownLatch.countDown();
                return albumInfoData.getUserId();
            }
        });

        executorService.execute(new Runnable() {
            @Override
            public void run() {
                //2.2 远程查询主播信息
                try {
                    Long userId = future.get();  //当前线程会阻塞 会等上一个线程运行完在执行
                    Result<UserInfoVo> albumInfoVoResult = userInfoFeignClient.getUserInfo(userId);
                    UserInfoVo userInfoVoData = albumInfoVoResult.getData();
                    Assert.notNull(userInfoVoData, "远程调用专辑微服务获取用户信息失败");
                    albumInfoIndex.setAnnouncerName(userInfoVoData.getNickname());

                } catch (Exception e) {
                    throw new RuntimeException(e);
                }finally {
                    countDownLatch.countDown();
                }

            }
        });


        executorService.execute(new Runnable() {
            @Override
            public void run() {
                //2.3 远程查询分类信息
                Result<BaseCategoryView> baseCategoryViewResult =  albumInfoFeignClient.getAlbumCategory(albumId);
                BaseCategoryView baseCategoryViewData = baseCategoryViewResult.getData();
                if (baseCategoryViewData == null){
                    throw new GuiguException(201,"远程调用专辑微服务获取专辑分类信息失败");
                }
                albumInfoIndex.setCategory1Id(baseCategoryViewData.getCategory1Id());
                albumInfoIndex.setCategory2Id(baseCategoryViewData.getCategory2Id());
                albumInfoIndex.setCategory3Id(baseCategoryViewData.getCategory3Id());
                countDownLatch.countDown();
            }
        });


        executorService.execute(new Runnable() {
            @Override
            public void run() {
                //2.4 远程查询统计信息
                Result<AlbumStatVo> albumStatVoResult =  albumInfoFeignClient.getAlbumStat(albumId);
                AlbumStatVo albumStatVoData = albumStatVoResult.getData();

                if (albumStatVoData == null){
                    throw new GuiguException(201,"远程调用专辑微服务获取专辑分类信息失败");
                }

                Integer playStatNum  = albumStatVoData.getPlayStatNum();
                Integer subscribeStatNum = albumStatVoData.getSubscribeStatNum();
                Integer buyStatNum = albumStatVoData.getBuyStatNum();
                Integer commentStatNum = albumStatVoData.getCommentStatNum();


                albumInfoIndex.setPlayStatNum(playStatNum);
                albumInfoIndex.setSubscribeStatNum(subscribeStatNum);
                albumInfoIndex.setBuyStatNum(buyStatNum);
                albumInfoIndex.setCommentStatNum(commentStatNum);


//        Double hotScore = playStatNum * 0.1 + subscribeStatNum * 0.2 + buyStatNum * 0.4 + commentStatNum * 0.3 ;//线上使用

                Double hotScore = new Random().nextDouble(); //测试使用

                albumInfoIndex.setHotScore(hotScore);
                countDownLatch.countDown();

            }
        });

        //3.将文档对象存储到es中

        countDownLatch.await();
        long endTime = System.currentTimeMillis();

        log.info("专辑：{}上架到es耗时：{}ms",albumId,endTime - startTime);
        albumInfoIndexRepository.save(albumInfoIndex);

        // 向suggestInfo索引库中保存数据

        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(albumInfoIndex.getId().toString());
        suggestIndex.setTitle(albumInfoIndex.getAlbumTitle());
        suggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumTitle()}));
        suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumTitle())}));
        suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumTitle())}));


        suggestIndexRepository.save(suggestIndex);


    }

    @Override
    public void albumOffSale(Long albumId) {
        try{
            albumInfoIndexRepository.deleteById(albumId);
            log.info("专辑下架成功");
        }catch (Exception e){
            log.error("专辑下架失败");
        }

    }
    @Override
    public void batchAlbumOffSale() {
        try {

            albumInfoIndexRepository.deleteAll();
            log.info("专辑批量下架成功");
        }catch (Exception e){
            log.error("专辑批量下架失败");
        }
    }

//    Map<Thread, String> flagMap = new HashMap<>();
    @SneakyThrows
    @Override
    public Map<String, Object> getAlbumInfo(Long albumId) {
//        return getDistroCacheAndLockFinallyRedisVersion(albumId);
//        return getDistroCacheAndLockFinallyRedissonVersion1(albumId);
//        return getDistroCacheAndLockFinallyRedissonVersion2(albumId);
        return getAlbumInfoFromDb(albumId);

    }

    private Map getDistroCacheAndLockFinallyRedissonVersion2(Long albumId) throws InterruptedException {
        //1.定义缓存key
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
        Long ttl = 0l;

        //2.查询分布式布隆过滤器
        boolean contains = rBloomFilter.contains(albumId);
        if (!contains){
            return null;
        }

        //3.查询缓存
        String jsonStrFromRedis = redisTemplate.opsForValue().get(cacheKey);

        //3.1缓存命中
        if (!StringUtils.isEmpty(jsonStrFromRedis)){
            return JSONObject.parseObject(jsonStrFromRedis,Map.class);
        }
        //3.2缓存未命中 查询数据库
        //3.2.1 添加分布式锁

        RLock lock = redissonClient.getLock(lockKey);
        boolean acquireLockFlag = lock.tryLock();
        if (acquireLockFlag){
            try {
                //3.2.2 回源查询数据
                HashMap<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);
                if (albumInfoFromDb != null ){
                    ttl = 60 * 60 * 24 * 7l;
                }else {
                    ttl = 60 * 60 * 24l;
                }
                //3.2.3 同步数据
                redisTemplate.opsForValue().set(cacheKey,JSONObject.toJSONString(albumInfoFromDb),ttl,TimeUnit.SECONDS);
                return albumInfoFromDb;

            }finally {
                lock.unlock();
            }

        }else {
            //等同步时间之后 查询缓存即可
            Thread.sleep(200);
            String result = redisTemplate.opsForValue().get(cacheKey);
            if (!StringUtils.isEmpty(result)){
                return JSONObject.parseObject(result, Map.class);
            }
            return getAlbumInfoFromDb(albumId);
        }
    }


    /**
     * 企业版本
     * @param albumId
     * @return
     * @throws InterruptedException
     */
    private Map getDistroCacheAndLockFinallyRedissonVersion1(Long albumId) throws InterruptedException {
        //1.定义缓存key
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
        Long ttl = 0l;

        //2.查询分布式布隆过滤器
        boolean contains = rBloomFilter.contains(albumId);
        if (!contains){
            return null;
        }

        //3.查询缓存
        String jsonStrFromRedis = redisTemplate.opsForValue().get(cacheKey);

        //3.1缓存命中
        if (!StringUtils.isEmpty(jsonStrFromRedis)){
            return JSONObject.parseObject(jsonStrFromRedis,Map.class);
        }
        //3.2缓存未命中 查询数据库
        //3.2.1 添加分布式锁

        RLock lock = redissonClient.getLock(lockKey);
        boolean acquireLockFlag = lock.tryLock();
        if (acquireLockFlag){
            try {
                //3.2.2 回源查询数据
                HashMap<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);
                if (albumInfoFromDb != null ){
                    ttl = 60 * 60 * 24 * 7l;
                }else {
                    ttl = 60 * 60 * 24l;
                }
                //3.2.3 同步数据
                redisTemplate.opsForValue().set(cacheKey,JSONObject.toJSONString(albumInfoFromDb),ttl,TimeUnit.SECONDS);
                return albumInfoFromDb;

            }finally {
                lock.unlock();
            }

        }else {
            //等同步时间之后 查询缓存即可
            Thread.sleep(200);
            String result = redisTemplate.opsForValue().get(cacheKey);
            if (!StringUtils.isEmpty(result)){
                return JSONObject.parseObject(result, Map.class);
            }
            return getAlbumInfoFromDb(albumId);
        }
    }




    /**
     * 面试版本
     * finally version
     * 缓存击穿(分布式锁) 缓存穿透（固定值攻击：缓存key-null  随机值攻击使用的是本地布隆过滤器）
     * Redis分布式锁 + 自旋可重入 + 双缓存查询的缓存使用
     * @param albumId
     * @return
     * 续期：
     * 要做的：只要抢到锁的线程没有把自己的活干完，这个抢到锁的线程对应的这个锁key就不能释放掉，只有等抢到锁的线程把活干完了或者干活的期间出异常了，才让锁过期
     * 活没干完 -- 给锁key续期
     * 活干完了或者期间出异常 --- 不用再给锁key续期
     * 注意：只有抢到锁才续期，没抢到就不续
     * 启动一个线程 --> 负责完成续期任务
     * 方案1：new Thread线程 让这个线程做续期任务（一直续） 并且让这个线程作为守护线程 最后再利用Thread的中断机制，完成对续期线程取消
     * 方案2：用线程池完成续期 [定时或者延时任务的线程池实现 ]
     *
     *      续期：每隔多久 再让redis中的锁key的时间是一个新值，每个10s 将redis中的锁key设置为30s
     */

    private Map getDistroCacheAndLockFinallyRedisVersion(Long albumId) {
        //1.定义变量
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
        Boolean acquireLockFlag = false;
        String token = "";

        //1.1 查询布隆过滤器(本地布隆过滤器)
//        boolean b = longBloomFilter.mightContain(albumId);  //解决缓存穿透的随机值攻击
//        if (!b){
//            return null;
//        }
        //1.1 查询布隆过滤器(分布式布隆过滤器)
        boolean bloomContains = rBloomFilter.contains(albumId);
        if (!bloomContains){
            return null;
        }

        //2.查询分布式缓存
        String resultFromCache = redisTemplate.opsForValue().get(cacheKey);

        //3.判断缓存是否存在
        if (!StringUtils.isEmpty(resultFromCache)){
            return JSONObject.parseObject(resultFromCache, Map.class);
        }

        //4.缓存未命中 准备查询数据库
        //4.1从ThreadLocal中获取令牌值（解决递归的线程进来）
        String s = reentrantLockTokenThreadLocal.get();
        //4.2如果是递归进来的线程
        if (StringUtils.isEmpty(s)){
           token = s;
            acquireLockFlag = true;
        }else {
            //4.3 第一次进来的
            token = UUID.randomUUID().toString().replace("-","");
            //4.4 加分布式锁
            acquireLockFlag = redisTemplate.opsForValue().setIfAbsent(lockKey, token,60,TimeUnit.SECONDS);
        }
        // 5.加分布式锁成功
        if (acquireLockFlag){

            //开始续期
            ExpireThreadExecutor expireThreadExecutor = new ExpireThreadExecutor(redisTemplate,albumId);
            expireThreadExecutor.renewal(30l,TimeUnit.SECONDS);
            Map<String, Object> albumInfoFromDb;
            try{
                long ttl = 0l;

                //5.1 回源查询数据库 //
                albumInfoFromDb = getAlbumInfoFromDb(albumId);  //map中有数据 --> 给该数据在redis中存储一个较长的时间
                                                                //map中没有数据 --> 给{}在redis中存储一个较短的时间
                if (albumInfoFromDb != null && albumInfoFromDb.size() > 0){
                    ttl = 60 * 60 * 24 * 7l;
                }else {
                    ttl = 60 * 60 * 2;
                }
                //5.2 将数据库查询的数据同步给缓存Redis
                redisTemplate.opsForValue().set(cacheKey,JSONObject.toJSONString(albumInfoFromDb),ttl,TimeUnit.SECONDS); //缓存穿透的固定值攻击解决
            }finally {
                //5.3释放锁
                //判断是不是自己加的锁  是 可以删除 不是 不能删除
                String luaScript = "if redis.call('get', KEYS[1]) == ARGV[1] then return redis.call('del', KEYS[1]) else return 0 end ";
                Long execute = redisTemplate.execute(new DefaultRedisScript<Long>(luaScript, Long.class), Arrays.asList(lockKey, token));
                if (execute == 0){
                    log.error("释放锁失败");
                }else {
                    log.info("释放锁成功");
                }
                //5.4 从ThreadLocal中移除令牌
                reentrantLockTokenThreadLocal.remove();  //防止内存泄露问题

                //5.5结束续期任务
                expireThreadExecutor.cancelRenewal();
            }
            //5.6 返回数据给前端
            return albumInfoFromDb;
        } else {
            //6.加分布式锁失败
            //6.1等同步时间
            try {
                Thread.sleep(200);  //一定要压测得到准确值
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            //6.2查询缓存 -- 正常99%的情况下200ms后缓存一定是有数据，所以直接返回出去
            String resultFromCacheStr = redisTemplate.opsForValue().get(cacheKey);
            if (!StringUtils.isEmpty(resultFromCacheStr)){
                return JSONObject.parseObject(resultFromCacheStr, Map.class);
            }

            //6.3 解决1%极端情况（抢到所得线程在将数据库中的数据同步到缓存的时候出现了问题，导致缓存没有）
            while (true){  //通过监控工具排查定位cpu飙升原因  //兜底  自旋就是while[自旋+可重复入]
                //6.4 查询缓存的作用：主要是为了解决，递归进去的线程将数据同步到缓存之后，其他线程还要抢锁
                String doubleCacheStr = redisTemplate.opsForValue().get(cacheKey);
                //6.5如果有，要在while true强所得线程不用再枪锁 直接将递归进去的线程放回到缓存中的数据返回即可
                if (!StringUtils.isEmpty(doubleCacheStr)){
                    return JSONObject.parseObject(doubleCacheStr, Map.class);
                }
                //6.6 抢锁
                Boolean acquireLock = redisTemplate.opsForValue().setIfAbsent(lockKey, token, 60, TimeUnit.SECONDS);
                if (acquireLock){
                    reentrantLockTokenThreadLocal.set(token); //将该线程枷锁的令牌存放到ThreadLocal，主要解决递归进去的线程不再加锁[保证可重入锁]
                    break;//退出循环
                }
            }
            //7. 重新递归进去查询数据库
            return getDistroCacheAndLockFinallyRedisVersion(albumId);
        }
    }

    /**
     * v3:解决锁的误删
     * 解决：每个线程都给一个锁标识，先判断一下锁是不是自己的  如果是则删除 如果不是 则不能删除
     * <p>
     * 问题：判断锁和释放锁不是原子操作
     * 解决办法：自定义lua脚本的表达式来将判断锁和释放锁做成原子操作，从而保证锁的误删真正被解决
     * @param albumId
     * @return
     */

    private  Map getDistroCacheAndLockV3(Long albumId) {
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
        String token = UUID.randomUUID().toString().replace("-","");
        //1.查询分布式缓存
        String resultFromCache = redisTemplate.opsForValue().get(cacheKey);

        //2.判断缓存是否存在
        if (!StringUtils.isEmpty(resultFromCache)) {
            return JSONObject.parseObject(resultFromCache, Map.class);
        }
        //3.缓存未命中 回源查询数据库
        //3.1 加分布式锁
        Boolean lock = redisTemplate.opsForValue().setIfAbsent(lockKey, token,30,TimeUnit.SECONDS);
        if (lock){
//            redisTemplate.expire(lockKey,30,TimeUnit.SECONDS); //服务端给锁一个过期时间 避免死锁发生
            //3.2 回源查询数据库
            Map<String, Object> albumInfoFromDb;
            try{
                albumInfoFromDb = getAlbumInfoFromDb(albumId);
                //3.3 将数据库查询的数据同步给缓存Redis
                redisTemplate.opsForValue().set(cacheKey,JSONObject.toJSONString(albumInfoFromDb));
            }finally {
                //3.4 释放锁
                //判断是不是自己加的锁  是 可以删除 不是 不能删除
                String luaScript = "IF redis.call('get',KEYS[1]) == ARGV[1] THEN return redis.call('del',KEYS[1]) ELSE return 0 END ";
                Long execute = redisTemplate.execute(new DefaultRedisScript<Long>(luaScript, Long.class), Arrays.asList(lockKey, token));
                if (execute == 0){
                    log.error("释放锁失败");
                }else {
                    log.info("释放锁成功");
                }
            }
            //3.5 返回数据给前端
            return albumInfoFromDb;
        }else {
            //4. 等同步时间
            try {
                Thread.sleep(200);  //一定要压测得到准确值
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String resultFromCacheStr = redisTemplate.opsForValue().get(cacheKey);
            return JSONObject.parseObject(resultFromCacheStr, Map.class);
        }
    }

    /**
     * v2:解决在极端情况再，抢到所得线程刚执行业务 断电 导致Redis中的锁keu没有被释放 那么就会导致死锁发生
     * 解决办法：
     * 1.客户端手动删除锁   RedisTemplate.delete(lockKey)
     * 2.redis服务端给锁key过期时间 redisTemplate.expire(lockKey,30,TimeUnit.SECONDS)
     * <p>
     * v2版本的问题
     *  在极端情况下：抢到锁的线程，正准备给锁key设置过期时间的时候断电了，那么还是会出现死锁问题
     *  解决办法：将加锁和给锁设置过期时间的两部动作做成一个原子操作，redisTemplate.opsForValue().setIfAbsent(lockKey,"lock",30,TimeUnit.SECONDS)
     *  底层实现：redis将设置key和给这个key设置过去时间用一个lua脚本包装起来，然后用一个redis链接执行这个lua脚本 从而保证这两个动作的原子性
     * @param albumId
     * @return
     */
    private @Nullable Map getDistroCacheAndLockV2(Long albumId) {
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
        //1.查询分布式缓存
        String resultFromCache = redisTemplate.opsForValue().get(cacheKey);

        //2.判断缓存是否存在
        if (!StringUtils.isEmpty(resultFromCache)){
            return JSONObject.parseObject(resultFromCache, Map.class);
        }

        //3.缓存未命中 回源查询数据库

        //3.1 加分布式锁
        Boolean lock = redisTemplate.opsForValue().setIfAbsent(lockKey, "lock",30,TimeUnit.SECONDS);
        if (lock){
//            redisTemplate.expire(lockKey,30,TimeUnit.SECONDS); //服务端给锁一个过期时间 避免死锁发生
            //3.2 回源查询数据库
            Map<String, Object> albumInfoFromDb;
            try{
                albumInfoFromDb = getAlbumInfoFromDb(albumId);
                //3.3 将数据库查询的数据同步给缓存Redis
                redisTemplate.opsForValue().set(cacheKey,JSONObject.toJSONString(albumInfoFromDb));
            }finally {
                //3.4 释放锁
                redisTemplate.delete(lockKey);
            }
            //3.5 返回数据给前端
            return albumInfoFromDb;

        }else {
            //4. 等同步时间
            try {
                Thread.sleep(200);  //一定要压测得到准确值
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String resultFromCacheStr = redisTemplate.opsForValue().get(cacheKey);
            return JSONObject.parseObject(resultFromCacheStr, Map.class);
        }
    }

    /**
     * 使用分布式缓存Redis + Redis分布式锁 解决缓存击穿问题
     */
    private @Nullable Map getDistroCacheAndLockV1(Long albumId) {
        String cacheKey = RedisConstant.CACHE_INFO_PREFIX + albumId;
        String lockKey = RedisConstant.ALBUM_LOCK_SUFFIX + albumId;
        //1.查询分布式缓存
        String resultFromCache = redisTemplate.opsForValue().get(cacheKey);

        //2.判断缓存是否存在
        if (!StringUtils.isEmpty(resultFromCache)){
            return JSONObject.parseObject(resultFromCache, Map.class);
        }

        //3.缓存未命中 回源查询数据库

        //3.1 加分布式锁
        Boolean lock = redisTemplate.opsForValue().setIfAbsent(lockKey, "lock");

        if (lock){
            //3.2 回源查询数据库
            Map<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);
            //3.3 将数据库查询的数据同步给缓存Redis
            redisTemplate.opsForValue().set(cacheKey,JSONObject.toJSONString(albumInfoFromDb));
            //3.4 释放锁
            redisTemplate.delete(lockKey);
            //3.5 返回数据给前端
            return albumInfoFromDb;
        }else {
            //4. 等同步时间
            try {
                Thread.sleep(200);  //一定要压测得到准确值
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            String resultFromCacheStr = redisTemplate.opsForValue().get(cacheKey);
            return JSONObject.parseObject(resultFromCacheStr, Map.class);
        }
    }


    /**
     * 使用双缓存加工业
     * 本地Map缓存 + 分布式Redis缓存
     * @param albumId
     * @return
     */
    /*@Override
    public Map<String, Object> getAlbumInfo(Long albumId) {
        Map<String,Object> result = new HashMap<>();

        Long startTime = System.currentTimeMillis();

        // 1.查询一级缓存(Map)
        if (localCache.containsKey(albumId)){
            log.info("本地缓存命中：整个接口的响应试件在：{}ms",System.currentTimeMillis() - startTime);
            return localCache.get(albumId);
        }
        log.info("本地l1缓存未命中，准备查询l2缓存：");

        // 2.查询二级缓存Redis
        String s = redisTemplate.opsForValue().get(albumId.toString());

        // 3.分布式缓存是否命中
        if (!StringUtils.isEmpty(s)){

            //将Redis的数据反序列化成为Map对象
            result = JSONObject.parseObject(s, Map.class);
        }else {
            // 4.分布式缓存未命中
            result = getAlbumInfoFromDb(albumId);
            // 5.将数据库查询的数据同步给Redis以及一级缓存
            redisTemplate.opsForValue().set(String.valueOf(albumId),JSONObject.toJSONString(result));
        }
        localCache.put(albumId,result);

        // 7.返回数据
        return result;

    }*/



    /**
     * 使用分布式缓存 + 异步组合
     * @param albumId
     * @return
     */
/*    @Override
    public Map<String, Object> getAlbumInfo(Long albumId) {
        long startTime = System.currentTimeMillis();

        //1.查询分布式缓存
        String s = redisTemplate.opsForValue().get(albumId.toString());
        //2.判断分布式缓存是否命中 ，命中直接返回给前端
        if (!StringUtils.isEmpty(s)){
            log.info("redis缓存命中，且整个接口的响应时间在：{}",System.currentTimeMillis() - startTime);
            return JSONObject.parseObject(s,Map.class);
        }
        log.info("分布式缓存未命中，开始会员操作");
        //3.分布式缓存未命中，回源查询数据库
        HashMap<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);

        //4.将数据库的数据同步到分布式缓存Redis中
        redisTemplate.opsForValue().set(albumId.toString(),JSONObject.toJSONString(albumInfoFromDb));

        log.info("分布式缓存未命中使用之后接口的响应时间在：{}",System.currentTimeMillis()-startTime);

        return albumInfoFromDb;


    }*/
    /**
     * 本地缓存+异步优化
     * @param albumId
     * @return
     */
  /*  @Override
    public Map<String, Object> getAlbumInfo(Long albumId) {

        long startTime = System.currentTimeMillis();


        //1.查询本地缓存
        boolean b = localCache.containsKey(albumId);
        //1.1 本地缓存命中，直接返回数据给前端
        if (b){
            log.info("本地缓存命中，且整个接口的响应时间在：{}",System.currentTimeMillis() - startTime);
            return localCache.get(albumId);
        }
        log.info("本地缓存未命中，开始回源操作");
        //1.2 本地缓存未命中 回源（查询数据库）
        HashMap<String, Object> albumInfoFromDb = getAlbumInfoFromDb(albumId);

        //2.将数据库的数据同步给本地缓存
        localCache.put(albumId,albumInfoFromDb);

        //3.将数据库查询的数据返回给前端
        long endTime = System.currentTimeMillis();
        log.info("使用本地缓存之后接口的响应时间在：{}",endTime-startTime);
        return albumInfoFromDb;

    }
*/
    @NotNull
    private HashMap<String, Object> getAlbumInfoFromDb(Long albumId) {
        // 1.创建Map对象
        HashMap<String, Object> map = new HashMap<>();

        CompletableFuture<Void> albumStatCompletableFuture = CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
                System.out.println("执行查询专辑统计任务用的线程" + Thread.currentThread().getName());
                // 1.专辑的统计信息
                Result<AlbumStatVo> albumStatVoResult = albumInfoFeignClient.getAlbumStat(albumId);
                AlbumStatVo albumStatVoData = albumStatVoResult.getData();
                if (albumStatVoData == null) {
                    throw new GuiguException(201, "远程查询专辑微服务专辑统计信息失败");
                }
                map.put("albumStatVo", albumStatVoData);
            }
        },threadPoolExecutor);

        CompletableFuture<Void> viewCompletableFuture = CompletableFuture.runAsync(new Runnable() {
            @Override
            public void run() {
                System.out.println("执行查询专辑分类任务用的线程" + Thread.currentThread().getName());
                // 2.专辑的分类（分类的名字）

                Result<BaseCategoryView> albumCategoryResult = albumInfoFeignClient.getAlbumCategory(albumId);
                BaseCategoryView baseCategoryViewData = albumCategoryResult.getData();
                if (baseCategoryViewData == null) {
                    throw new GuiguException(201, "远程查询专辑微服务专辑分类信息失败");
                }

                map.put("baseCategoryView", baseCategoryViewData);
            }
        },threadPoolExecutor);
        CompletableFuture<Long> albumInfoCompletableFuture = CompletableFuture.supplyAsync(new Supplier<Long>() {
            @Override
            public Long get() {
                System.out.println("执行查询专辑基本任务用的线程" + Thread.currentThread().getName());
                // 3.专辑基本数据
                Result<AlbumInfo> albumInfoAndAttrValueResult = albumInfoFeignClient.getAlbumInfoAndAttrValue(albumId);
                AlbumInfo albumInfoData = albumInfoAndAttrValueResult.getData();
                if (albumInfoData == null) {
                    throw new GuiguException(201, "远程查询专辑微服务专辑基本信息失败");
                }
                map.put("albumInfo", albumInfoData);
                return albumInfoData.getUserId();
            }
        }, threadPoolExecutor);
        CompletableFuture<Void> userCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(new Consumer<Long>() {
            @Override
            public void accept(Long userId) {
                System.out.println("执行查询专辑主播任务用的线程" + Thread.currentThread().getName());
                // 4.专辑的主播信息
                Result<UserInfoVo> userInfoResult = userInfoFeignClient.getUserInfo(userId);
                UserInfoVo userInfoResultData = userInfoResult.getData();
                if (userInfoResultData == null){
                    throw new GuiguException(201,"远程查询用户微服务专辑主播信息失败");
                }
                map.put("announcer",userInfoResultData);
            }
        },threadPoolExecutor);

        CompletableFuture.allOf(albumStatCompletableFuture,viewCompletableFuture,albumInfoCompletableFuture,userCompletableFuture).join();

        return map;
    }

    @SneakyThrows
    @Override
    public void preRankingCache() {
        //1.查询全平台的一级分类id
        Result<List<Long>> category1IdsResult = albumInfoFeignClient.getAllCategory1Id();
        List<Long> c1IdData = category1IdsResult.getData();
        if (CollectionUtils.isEmpty(c1IdData)){
            throw new GuiguException(201,"远程查询专辑微服务获取一级分类id失败");
        }

        for (Long c1Id : c1IdData) {
            String[] fiveDimension = {"hotScore","playStatNum","subscribeStatNum","buyStatNum","commentStatNum"};
            for (String dimension : fiveDimension) {
                SearchResponse<AlbumInfoIndex> response = elasticsearchClient.search(srb -> srb
                        .index("albuminfo")
                        .query(qb -> qb
                                .term(tqb -> tqb
                                        .field("category1Id")
                                        .value(c1Id)))
                        .sort(sob -> sob
                                .field(fsb -> fsb
                                        .field(dimension)))
                        .size(10), AlbumInfoIndex.class);

                List<AlbumInfoIndex> albumInfoIndices = new ArrayList<>();

                for (Hit<AlbumInfoIndex> hit : response.hits().hits()) {
                    AlbumInfoIndex albumInfoIndex = hit.source();
                    albumInfoIndices.add(albumInfoIndex);
                }

                String bigKey = RedisConstant.RANKING_KEY_PREFIX + c1Id;

                redisTemplate.opsForHash().put(bigKey,dimension, JSONObject.toJSONString(albumInfoIndices));
            }
        }

    }

    @Override
    public List<AlbumInfoIndexVo> findRankingList(Long category1Id, String dimension) {
        String bigKey = RedisConstant.RANKING_KEY_PREFIX + category1Id;
        String albumInfoIndexList  = (String) redisTemplate.opsForHash().get(bigKey, dimension);

        if (StringUtils.isEmpty(albumInfoIndexList)){
            throw new GuiguException(201,"排行榜信息不存在");
        }
        List<AlbumInfoIndex> albumInfoIndices = JSONObject.parseArray(albumInfoIndexList, AlbumInfoIndex.class);
        List<AlbumInfoIndexVo> albumInfoIndexVoList = albumInfoIndices.stream().map(albumInfoIndex -> {
            AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
            BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
            return albumInfoIndexVo;
        }).collect(Collectors.toList());

        return albumInfoIndexVoList;
    }

    @Override
    public List<Long> getAlbumInfoIdList() {
        Result<List<Long>> albumIds = albumInfoFeignClient.getAlbumInfoIdList();
        List<Long> albumIdsData = albumIds.getData();
        if (CollectionUtils.isEmpty(albumIdsData)){
            throw new GuiguException(201,"应用中不存在专辑id集合");
        }
        return albumIdsData;
    }

    @Override
    public Boolean rebuildBloomFilter() {
        //todo 如何创建布隆
        /*
        做法：
            1.删除老布隆
            2.删除老布隆的配置
            3.创建新布隆
            4.初始化新布隆
            5.将数据放到新布隆
            6.新布隆商检就可以使用
         */
        /*
        优化做法：
            1.创建新布隆
            2.初始化新布隆
            3，将数据放到新布隆
            4.删除老布隆的数据
            5.删除老布隆的配置
            6.重新用：劳布龙的名字换新布隆的名字 -- 456做成一个原子操作

            新布隆上线就可以使用 rename：重新命名 albumInfoBloomFilterName albumInfoBloomFilter
         */
        RBloomFilter<Object> albumIdBloomFilterNew = redissonClient.getBloomFilter("albumIdBloomFilterNew");
        albumIdBloomFilterNew.tryInit(1000000l, 0.001);

        List<Long> albumInfoIdList = getAlbumInfoIdList();
        for (Long albumId : albumInfoIdList) {
            albumIdBloomFilterNew.add(albumId);
        }

        //rename key key1 用key1的名字换key的名字
        String script = " redis.call(\"del\",KEYS[1])" +
                "  redis.call(\"del\",KEYS[2])" +
                "  redis.call(\"rename\",KEYS[3],KEYS[1])" +
                "  redis.call(\"rename\",KEYS[4],KEYS[2]) return 0";
        List<String> asList = Arrays.asList("albumIdBloomFilter", "{albumIdBloomFilter}:config", "albumIdBloomFilterNew", "{albumIdBloomFilterNew}:config");
        Long execute = redisTemplate.execute(new DefaultRedisScript<Long>(script, Long.class), asList);
        if (execute == 0){
            log.info("老布隆已经被删除，新布隆上线。。。");
        }
        return execute == 0;
    }


//    @Override
//    public void albumOnSale(Long albumId) {
//        //1.创建文档对象
//        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
//
//
//
//        //2.给albumInfoIndex属性赋值
//
//        Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfoAndAttrValue(albumId);
//
//        AlbumInfo albumInfoData = albumInfoResult.getData();
//        if (albumInfoData == null){
//            throw new GuiguException(201,"远程调用专辑微服务获取专辑信息失败");
//        }
//
//        albumInfoIndex.setId(albumInfoData.getId());
//        albumInfoIndex.setAlbumTitle(albumInfoData.getAlbumTitle());
//        albumInfoIndex.setAlbumIntro(albumInfoData.getAlbumIntro());
//
//        albumInfoIndex.setCoverUrl(albumInfoData.getCoverUrl());
//        albumInfoIndex.setIncludeTrackCount(albumInfoData.getIncludeTrackCount());
//        albumInfoIndex.setIsFinished(albumInfoData.getIsFinished().toString());
//        albumInfoIndex.setPayType(albumInfoData.getPayType());
//        albumInfoIndex.setCreateTime(new Date());
//
//        List<AttributeValueIndex> attributeValueIndexs = albumInfoData.getAlbumAttributeValueVoList().stream()
//                .map(albumAttributeValue -> {
//                    AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
//                    attributeValueIndex.setAttributeId(albumAttributeValue.getAttributeId());
//                    attributeValueIndex.setValueId(albumAttributeValue.getValueId());
//                    return attributeValueIndex;
//
//        }).collect(Collectors.toList());
//
//
//
//        albumInfoIndex.setAttributeValueIndexList(Lists.newArrayList(attributeValueIndexs));
//
//        Result<UserInfoVo> albumInfoVoResult = userInfoFeignClient.getUserInfo(albumInfoData.getUserId());
//        UserInfoVo userInfoVoData = albumInfoVoResult.getData();
//        Assert.notNull(userInfoVoData,"远程调用专辑微服务获取用户信息失败");
//        albumInfoIndex.setAnnouncerName(userInfoVoData.getNickname());
//
//        Result<BaseCategoryView> baseCategoryViewResult =  albumInfoFeignClient.getAlbumCategory(albumId);
//        BaseCategoryView baseCategoryViewData = baseCategoryViewResult.getData();
//        Assert.notNull(baseCategoryViewData,"远程调用专辑微服务分类信息失败");
//        albumInfoIndex.setCategory1Id(baseCategoryViewData.getCategory1Id());
//        albumInfoIndex.setCategory2Id(baseCategoryViewData.getCategory2Id());
//        albumInfoIndex.setCategory3Id(baseCategoryViewData.getCategory3Id());
//
//
//        Result<AlbumStatVo> albumStatVoResult =  albumInfoFeignClient.getAlbumStat(albumId);
//        AlbumStatVo albumStatVoResultData = albumStatVoResult.getData();
//
//        Assert.notNull(baseCategoryViewData,"远程调用专辑微服务专辑统计信息失败");
//
//        Integer playStatNum  = albumStatVoResultData.getPlayStatNum();
//        Integer subscribeStatNum = albumStatVoResultData.getSubscribeStatNum();
//        Integer buyStatNum = albumStatVoResultData.getBuyStatNum();
//        Integer commentStatNum = albumStatVoResultData.getCommentStatNum();
//
//
//        albumInfoIndex.setPlayStatNum(playStatNum);
//        albumInfoIndex.setSubscribeStatNum(subscribeStatNum);
//        albumInfoIndex.setBuyStatNum(buyStatNum);
//        albumInfoIndex.setCommentStatNum(commentStatNum);
//
//
////        Double hotScore = playStatNum * 0.1 + subscribeStatNum * 0.2 + buyStatNum * 0.4 + commentStatNum * 0.3 ;//线上使用
//
//        Double hotScore = new Random().nextDouble(); //测试使用
//
//
//        albumInfoIndex.setHotScore(hotScore);
//
//
//        //3.将文档对象存储到es中
//        albumInfoIndexRepository.save(albumInfoIndex);
//    }
}
